package cn.bby.query.server.infrastructure.listener;

import cn.bby.query.server.application.manager.DataChangeEventManager;
import cn.bby.query.server.domain.agg.metadata.entity.DsSettingsInfo;
import cn.bby.query.server.domain.agg.metadata.repository.DsSettingsRepository;
import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson2.JSON;
import io.debezium.connector.postgresql.PostgresConnector;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * @author ben
 * @since 2025/4/26 14:46
 */
@Slf4j
//@Component
@RequiredArgsConstructor
public class DebeziumDataListener implements DataListener {

    private final DataChangeEventManager dataChangeEventManager;
    private final DsSettingsRepository dsSettingsRepository;
    private final RedissonClient redissonClient;
    private boolean ready = false;

    @Override
    public void init() {
        ready = true;
        checkAndRun();
    }

    public void checkAndRun() {
        if (!ready) {
            return;
        }
        List<DsSettingsInfo> tableInfos = dsSettingsRepository.getAllTableInfos();
        if (tableInfos.isEmpty()) {
            return;
        }
        Map<Long, List<DsSettingsInfo>> dbMap = tableInfos.stream()
                .collect(Collectors.groupingBy(DsSettingsInfo::getDbId));
        dbMap.values().forEach(this::tryStart);
    }

    private void tryStart(List<DsSettingsInfo> tableInfos) {
        DsSettingsInfo dbInfo = CollUtil.getFirst(tableInfos);
        if (dbInfo == null) {
            return;
        }
        List<String> schemaTables = tableInfos.stream()
                .map(DsSettingsInfo::getTableName)
                .collect(Collectors.toList());
        Properties props = new Properties();
        props.setProperty("name", "engine-" + dbInfo.getDbName() + "-" + dbInfo.getDbId());
        props.setProperty("connector.class", PostgresConnector.class.getName());
        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        File file = new File("d:/tmp");
        if (!file.exists()) {
            file.mkdirs();
        }
        props.setProperty("offset.storage.file.filename", "d:/tmp/offsets.dat");  // 偏移量存储路径
        props.setProperty("offset.flush.interval.ms", "60000");
        props.setProperty("database.hostname", dbInfo.getHost());
        props.setProperty("database.port", dbInfo.getPort());
        props.setProperty("plugin.name", "pgoutput");
        props.setProperty("database.dbname", dbInfo.getDbName());
        props.setProperty("database.user", dbInfo.getUsername());
        props.setProperty("database.password", dbInfo.getPassword());
        props.setProperty("database.server.id", dbInfo.getDbId().toString()); // 唯一ID
        props.setProperty("topic.prefix", "topic-" + dbInfo.getDbName() + "-" + dbInfo.getDbId());
        props.setProperty("database.dbname", dbInfo.getDbName()); // 监控的数据库
        props.setProperty("name.include.list", String.join(",", schemaTables));  // 监控的数据表
        props.setProperty("snapshot.mode", "no_data"); // 不执行全量快照
        // 创建Debezium引擎
        DebeziumEngine.create(Json.class)
                .using(props)
                .notifying((records, committer) -> {
                    List<String> payloads = new ArrayList<>(records.size());
                    // 处理变更事件
                    for (ChangeEvent<String, String> record : records) {
                        try {
                            String val = record.value();
                            int idx = val.indexOf("\"payload\":{");
                            String payload = idx > 10 ? val.substring(idx + 10, val.length() - 1) : JSON.parseObject(val, ValueEntity.class).getPayload();
                            payloads.add(payload);
                        } catch (Exception e) {
                            log.debug("record:{}", record);
                            log.error("数据转换异常", e);
                        }
                        committer.markProcessed(record);
                    }
                    try {
                        dataChangeEventManager.sendEvent(payloads);
                    } catch (Exception e) {
                        log.error("数据写入异常", e);
                    }
                }).build().run();
    }

    @Data
    public static class ValueEntity {
        public String payload;
    }

}
